insert overwrite table jms_dm.dm_forecast_active_send_arrival_hours_detail_dt partition(dt='{{ execution_date | cst_ds }}')
select
    send_site_code          --发件站点
     ,can_arrival_site_code --到件站点
     ,duration_hours        --耗时（小时）
     ,send_scan_hm          --发件是分
     ,waybill_cnt
from
    (
        select
            send_site_code
             ,can_arrival_site_code
             ,duration_hours
             ,send_scan_hm
             ,waybill_cnt
             --根据当前站点到下游站点每个是小时单量最大的花费时间
             ,ROW_NUMBER() over(partition by send_site_code ,can_arrival_site_code,send_scan_hm order by waybill_cnt desc) as rk
        from
            (
                select
                    send_site_code
                     ,can_arrival_site_code
                     ,duration_hours
                     ,send_scan_hm
                     ,count(1) as waybill_cnt
                from
                    (  --例如A,[轨迹序号_B_发件时间,c]
                        --==> A ,B,B到件时间-A发件时间
                        --==> A ,c,c到件时间-A发件时间
                        select
                            waybill_no                                          --单号
                             ,send_site_code                                     --发件时间
                             ,send_scan_time                                     --发件扫描时间
                             ,date_format(send_scan_time,'HH:00:00') as send_scan_hm --发件时间所属小时
                             ,send_station_number                                --当前发件站点序号
                             ,send_site_type                                     --当前站点发件类型
                             ,index2
                             ,data2
                             ,split(data2,'_')[0] as scan_send_station_number    --下游网点B\c轨迹序号
                             ,split(data2,'_')[1] as can_arrival_site_code       --下游网点B\c
                             ,split(data2,'_')[2] as can_arrival_site_scan_time  --下游网点B\c到件时间
                             --当前站点到下游站点B\c花费时间(小时)
                             ,ceiling((unix_timestamp(split(data2,'_')[2],'yyyy-MM-dd HH:mm:ss') -unix_timestamp(send_scan_time,'yyyy-MM-dd HH:mm:ss'))/60/60 ) as duration_hours
                             --当前单号扫描是否有轨迹异常
                             ,if((min(nvl((unix_timestamp(split(data2,'_')[2],'yyyy-MM-dd HH:mm:ss') -unix_timestamp(send_scan_time,'yyyy-MM-dd HH:mm:ss')) ,-1))
                            over(partition by waybill_no) )<0,0,1) as is_need
                        from
                            (   --取最近3天的数据实际轨迹数数据
                                --实际轨迹 a-A-B-c ==>a,[轨迹序号_A_发件时间,B,c]，A,[轨迹序号_B_发件时间,c]，B,[轨迹序号_c_发件时间]
                                select waybill_no
                                     ,send_station_network_code    as send_site_code      --发件网点
                                     ,send_station_send_later_time as  send_scan_time    --发件时间
                                     ,send_station_arrival_early_time  as  arrival_scan_time  --到件站扫描时间
                                     ,send_station_number  --序号
                                     ,send_station_network_type    as send_site_type     --到件站点类型 
                                     --当前发件站点轨迹之后的站点数组
                                     ,sort_array(collect_list(concat(lpad(send_station_number,2,'0'),'_',send_station_network_code,'_',send_station_arrival_early_time))
                                    over(partition by waybill_no order by send_station_number ROWS BETWEEN 1 FOLLOWING AND  UNBOUNDED FOLLOWING )) as trace_codes_info
                                from jms_dwd.dwd_wide_sign_trace_waybill_step_dt
                                where dt between date_add('{{ execution_date | cst_ds }}',-3) and date_add('{{ execution_date | cst_ds }}',-0)
                                  and send_station_send_later_time is not null
                                  --and send_station_arrival_early_time is not null
                                  and is_refund=0
                                  and is_delete=1
                                --and waybill_no='JT2060949042771'
                            )t lateral view posexplode(trace_codes_info) view1 as index2,data2
                    )tt where is_need =1  --剔除有轨迹异常(运单轨迹中有下一站的发件时间小于上一站的发件时间的单号)
                group by send_site_code
                       ,can_arrival_site_code
                       ,duration_hours
                       ,send_scan_hm
            )ttt
    )tttt where rk=1
    distribute by hash(concat(send_site_code,can_arrival_site_code,send_scan_hm))%40
;